package com.hgcode.platform.rocketmq.consumer;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.hgcode.platform.config.rocketmq.RocketMqConfig;
import com.hgcode.platform.entity.User;
import com.hgcode.platform.repository.UserMapper;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * 功能描述: 同步MDM数据
 *
 * @auther: wentao_tang
 * @date: 2018-11-27 16:07
 */
@Component
public class OrderConsumer implements MessageListenerConcurrently {

    private static final Logger logger = LoggerFactory.getLogger(OrderConsumer.class);

    @Autowired
    private RocketMqConfig rocketMqConfig;

    private DefaultMQPushConsumer consumer =null;

    @Resource
    private UserMapper userMapper;

    @PostConstruct
    public void start() {
        try {
            consumer = new DefaultMQPushConsumer("12121212");

            logger.info("mq-order-consume starting ......");
            consumer.setNamesrvAddr(rocketMqConfig.getNamesrvAddr());
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.setMessageModel(MessageModel.CLUSTERING);
            consumer.subscribe(rocketMqConfig.getOrderTopic(), rocketMqConfig.getOrderTag());
            consumer.registerMessageListener(this);
            //失败重新消费的次数 不包括第一次
            consumer.setMaxReconsumeTimes(3);
            //最大默认64
            consumer.setConsumeThreadMax(1);
            // 最小默认20
            consumer.setConsumeThreadMin(1);
            consumer.start();
            logger.info("mq-order-consume start is success ! group:{},topic:{},tag:{}",
                rocketMqConfig.getOrderGroup(), rocketMqConfig.getOrderTopic(), rocketMqConfig.getOrderGroup());
        } catch (MQClientException e) {
            logger.error("mq-order-consume start is fail", e.getResponseCode(), e.getErrorMessage());
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @PreDestroy
    public void stop() {
        if (consumer != null) {
            consumer.shutdown();
            logger.info("MQ：stop permission_sync_data success! ");
        }
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        try {
            Long current=System.currentTimeMillis();
            for (int index = 0; index < msgList.size(); index++) {
                String messageBody = "";

                MessageExt msg = msgList.get(index);
                try {
                    messageBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
                if (StringUtils.isBlank(messageBody)) {
                    continue;
                }
                User content= JSON.parseObject(messageBody, User.class);
                if(content!=null){
                   User temp= userMapper.findById(content.getId());
                   if(temp==null){
                       userMapper.insertMe(content);
                   }
                }
                logger.info("order-consumer-message : body is:{}", messageBody);
            }
            logger.info("耗时：{}毫秒",System.currentTimeMillis()-current);
        }catch (Exception e){
            logger.warn("mdm sync data error:{}", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
